package com.zy.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import com.zy.gmall.realtime.bean.TrafficPageViewBean;
import com.zy.gmall.realtime.util.DateFormatUtil;
import com.zy.gmall.realtime.util.KafkaUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.doris.flink.cfg.DorisExecutionOptions;
import org.apache.doris.flink.cfg.DorisOptions;
import org.apache.doris.flink.cfg.DorisReadOptions;
import org.apache.doris.flink.sink.DorisSink;
import org.apache.doris.flink.sink.writer.SimpleStringSerializer;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.TimestampAssigner;
import org.apache.flink.api.common.eventtime.TimestampAssignerSupplier;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.runtime.rest.handler.job.savepoints.SavepointHandlers;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.Properties;

//版本，地区，渠道，新老访客聚合统计
public class DwsTrafficVcChArIsNewPageViewWindow{
    public static void main(String[] args) {
        //1
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        //2
        env.enableCheckpointing(500000L);
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,3000L));

        //3
        String topic = "topic_pages";
        String groupId = "dws_traffic_vc_ch_ar_isnew_group";
        KafkaSource<String> kafkaSource = KafkaUtil.getKafkaSource(topic, groupId);
        DataStreamSource<String> noWaterDS = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafkaSource");//获取数据源但是没指定水位线
        //noWaterDS.print(">>>>>>");
        //>>>>>>:3> {"common":{"ar":"19","uid":"77","os":"iOS 13.3.1","ch":"Appstore","is_new":"1","md":"iPhone 14 Plus","mid":"mid_456","vc":"v2.1.132","ba":"iPhone","sid":"76a6aefc-3f5c-47ab-abb1-72bd0942c33d"},
        // "page":{"page_id":"home","refer_id":"3","during_time":7723},
        // "ts":1654771809000}


        //4
        SingleOutputStreamOperator<JSONObject> mapDs = noWaterDS.map(new MapFunction<String, JSONObject>() {
            @Override
            public JSONObject map(String s) throws Exception {
                return JSON.parseObject(s);
            }
        });
        mapDs.print(">>>>>>>");

        //5 按照mid keyBy
        KeyedStream<JSONObject, String> keyByDs = mapDs.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject jsonObject) throws Exception {
                return jsonObject.getJSONObject("common").getString("mid");
            }
        });

        //6 数据转换 实体类 并筛选出 是否是首次登陆 是否是单日首次登录
        SingleOutputStreamOperator<TrafficPageViewBean> transformProcess = keyByDs.process(
                new KeyedProcessFunction<String, JSONObject, TrafficPageViewBean>() {

                    private ValueState<String> lastVisitDateState;//只用到了一个
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> lastVisitDateStateDescriptor
                                = new ValueStateDescriptor<String>("lastVisitDateState", String.class);
                        lastVisitDateStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());//独立访客每日重置，所以状态最多维护一天即可，为了减少内存开销，设置状态ttl为1天。
                        lastVisitDateState = getRuntimeContext().getState(lastVisitDateStateDescriptor);

                    }
                    @Override
                    public void processElement(JSONObject jsonObject, KeyedProcessFunction<String, JSONObject, TrafficPageViewBean>.Context context, Collector<TrafficPageViewBean> collector) throws Exception {
                        //判断 last_page_id 是否为 null，如果是，说明页面是首页，开启了一个新的会话，将会话数置为 1，否则置为 0。
                        JSONObject page = jsonObject.getJSONObject("page");
                        String durTime = page.getString("during_time");
                        JSONObject common = jsonObject.getJSONObject("common");
                        String ar = common.getString("ar");
                        String ch = common.getString("ch");
                        String isNew = common.getString("is_new");
                        String vc = common.getString("vc");
                        Long ts = jsonObject.getLong("ts");

                        String VisitTime = DateFormatUtil.toYmdHms(ts);
                        Long svCt = 0L;//会话
                        Long uvCt = 0L;//独立访客

                        if (StringUtils.isEmpty(page.getString("last_page_id"))) {
                            svCt = 1L;
                        }

                        String value = lastVisitDateState.value();//获取状态
                        // 最后在状态中维护末次访问日期，如果状态中的日期为null或不为当日，则将独立访客数置为1，否则置为0。
                        if (StringUtils.isEmpty(value) || !value.equals(VisitTime)) {
                            uvCt = 1L;
                            lastVisitDateState.update(VisitTime);
                        }



                        collector.collect(new TrafficPageViewBean(
                                "",
                                "",
                                vc,
                                ch,
                                ar,
                                isNew,
                                "",
                                uvCt,
                                svCt,
                                1L,
                                page.getLong("during_time"),//null
                                ts
                        ));

                    }
                }
        );
        //transforProcess.print(">>>>>>");

        //7 获取水位线
        SingleOutputStreamOperator<TrafficPageViewBean> watermarksDs = transformProcess.assignTimestampsAndWatermarks(
                WatermarkStrategy.<TrafficPageViewBean>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<TrafficPageViewBean>() {
                            @Override
                            public long extractTimestamp(TrafficPageViewBean trafficPageViewBean, long l) {
                                return trafficPageViewBean.getTs();
                            }
                        })
        );

        //8 分组 根据维度分区
        KeyedStream<TrafficPageViewBean, Tuple4<String, String, String, String>> keyedStreamDs = watermarksDs.keyBy(
                new KeySelector<TrafficPageViewBean, Tuple4<String, String, String, String>>() {
                    @Override
                    public Tuple4<String, String, String, String> getKey(TrafficPageViewBean trafficPageViewBean) throws Exception {
                        //vc,ch,ar,isNew
                        return Tuple4.of(trafficPageViewBean.getVc(), trafficPageViewBean.getCh(), trafficPageViewBean.getAr(), trafficPageViewBean.getIsNew());
                    }
        });
        //keyedStreamDs.print(">>>>>");


        //9 开窗
        WindowedStream<TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow> windowDs
                = keyedStreamDs.window(
                TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(5))
        );

        //测试找到是数据源的问题，没有过滤完全start
        //10 聚合
        SingleOutputStreamOperator<TrafficPageViewBean> reduceDs = windowDs.reduce(
                new ReduceFunction<TrafficPageViewBean>() {
                    @Override
                    public TrafficPageViewBean reduce(TrafficPageViewBean t2, TrafficPageViewBean t1) throws Exception {
                        t2.setPvCt(t2.getPvCt() + t1.getPvCt());
                        t2.setDurSum(t1.getDurSum() + t2.getDurSum());
                        t2.setSvCt(t1.getSvCt() + t2.getSvCt());
                        t2.setUvCt(t1.getUvCt() + t2.getUvCt());
                        return t2;
                    }
                },
                //补充数据
                new WindowFunction<TrafficPageViewBean, TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow>() {
                    @Override
                    public void apply(Tuple4<String, String, String, String> stringStringStringStringTuple4, TimeWindow timeWindow, Iterable<TrafficPageViewBean> iterable, Collector<TrafficPageViewBean> collector) throws Exception {
                        String startToYDM = DateFormatUtil.toYmdHms(timeWindow.getStart());
                        String endToYDM = DateFormatUtil.toYmdHms(timeWindow.getEnd());
                        String startToDate = DateFormatUtil.toDate(timeWindow.getStart());

                        for (TrafficPageViewBean trafficPageViewBean : iterable) {
                            trafficPageViewBean.setStt(startToYDM);
                            trafficPageViewBean.setEdt(endToYDM);
                            trafficPageViewBean.setCurDate(startToDate);
                            //数据没有导向下游
                            collector.collect(trafficPageViewBean);
                        }
                    }
                }
        );

        //reduceDs.print(">>>>>>>");
        // 11 输出到doris
        //输出语句提取
        Properties properties = new Properties();
        properties.setProperty("format","json");
        properties.setProperty("read_json_by_line","true");

        DorisSink<String> dorisSink = DorisSink.<String>builder()
                .setDorisReadOptions(DorisReadOptions.builder().build())
                .setDorisOptions(DorisOptions.builder()
                        .setFenodes("hadoop102:7030")
                        .setTableIdentifier("gmall.dws_traffic_vc_ch_ar_is_new_page_view_window")
                        .setUsername("root")
                        .setPassword("000000")
                        .build())
                .setDorisExecutionOptions(DorisExecutionOptions.builder()
                        .setLabelPrefix("label-doris") //streamload label prefix
                        .disable2PC() // 开启两阶段提交后,labelPrefix 需要全局唯一,为了测试方便禁用两阶段提交
                        .setDeletable(false)
                        .setBufferCount(3) // 批次条数: 默认 3
                        .setBufferSize(8 * 1024) // 批次大小: 默认 1M
                        .setCheckInterval(3000) // 批次输出间隔   三个对批次的限制是或的关系
                        .setMaxRetries(3)
                        .setStreamLoadProp(properties) // 设置 stream load 的数据格式 默认是 csv,需要改成 json
                        .build())
                .setSerializer(new SimpleStringSerializer())
                .build();

        //转换提取
        reduceDs.map(new MapFunction<TrafficPageViewBean, String>() {
            //转换成JSONString
            @Override
            public String map(TrafficPageViewBean trafficPageViewBean) throws Exception {
                SerializeConfig config = new SerializeConfig();
                config.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;
                return JSON.toJSONString(trafficPageViewBean, config);
            }
        }).sinkTo(dorisSink);

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
